package com.suncreate.atm.getData;

import cn.hutool.core.convert.Convert;
import cn.hutool.core.date.DateUtil;
import cn.hutool.core.util.ObjUtil;
import cn.hutool.core.util.StrUtil;
import cn.hutool.db.Db;
import cn.hutool.db.Entity;
import cn.hutool.db.handler.EntityListHandler;
import cn.hutool.db.sql.SqlBuilder;
import cn.hutool.log.StaticLog;
import cn.hutool.setting.Setting;

import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.time.LocalDateTime;
import java.util.List;
import com.alibaba.druid.pool.DruidDataSource;
import com.mysql.cj.util.StringUtils;
import com.suncreate.atm.utils.DataBasesUtils;

public class MySQLSync {

    // 加载配置文件
    static Setting setting = new Setting("jdbc.properties");

    // 源数据库和目标数据库配置
    static String sourceUrl = setting.get("from.jdbc.url");
    static String sourceUsername = setting.get("from.jdbc.user");
    static String sourcePassword = setting.get("from.jdbc.password");

    static String targetUrl = setting.get("to.jdbc.url");
    static String targetUsername = setting.get("to.jdbc.user");
    static String targetPassword = setting.get("to.jdbc.password");

    static String fullSyncTables = setting.get("fullSyncTables");
    static String addedSyncTables = setting.get("addedSyncTables");
    static String addedTimeColumName = setting.get("addedTimeColumName");

    static int batchSize = Convert.toInt(setting.get("batchSize"));
    //判断当前APP是否在运行
    static String thisAppName = setting.get("thisAppName");

    public static void run() {

        DruidDataSource sourceDataSource = null;
        DruidDataSource targetDataSource = null;

        try {
            // 初始化数据源
            sourceDataSource = DataBasesUtils.createDataSource(sourceUrl, sourceUsername, sourcePassword);
            targetDataSource = DataBasesUtils.createDataSource(targetUrl, targetUsername, targetPassword);

            // 初始化数据库连接
            Db sourceDb = Db.use(sourceDataSource);
            Db targetDb = Db.use(targetDataSource);

            /********** 下为【增量同步】部分 *******/
            StaticLog.info("\n");
            StaticLog.info("【增量】同步开始 ...");

            // 增量同步
            long startTimeAdded = System.currentTimeMillis();

            // 读取上次同步时间
            String nowTimeStr = DateUtil.now();
            for (String tableNames : StrUtil.split(addedSyncTables, ',')) {
                 incrementalSync(sourceDb, targetDb,tableNames,addedTimeColumName,nowTimeStr);
            }

            long endTimeAdded = System.currentTimeMillis();
            long durationAdded=(endTimeAdded - startTimeAdded)/1000;
            StaticLog.info("【增量】同步完成，耗时: {} 秒 \n", durationAdded);

            /********** 下为【全量同步】部分 *******/
            StaticLog.info("\n");
            StaticLog.info("【全量】同步开始 ...");
            long startTimeFull = System.currentTimeMillis();
            // 全量同步
            for (String tableNames : StrUtil.split(fullSyncTables, ',')) {
                fullSync(sourceDb, targetDb, tableNames);
            }
            long endTimeFull = System.currentTimeMillis();
            long durationFull=(endTimeFull - startTimeFull)/1000;

            StaticLog.info("【全量】同步完成，耗时: {} 秒 \n", durationFull);

            StaticLog.info("【增量+全量】同步完成，耗时: {} 秒 \n", durationFull+durationAdded);

        } catch (SQLException e) {
            e.printStackTrace();
        }
        finally {
            // 关闭数据库连接池
            DataBasesUtils.closeTheConnection(sourceDataSource);
            DataBasesUtils.closeTheConnection(targetDataSource);
            StaticLog.info("数据库连接已关闭。");
        }
    }


    /**
     * 全量同步
     */
    private static void fullSync(Db sourceDb, Db targetDb, String tableName) throws SQLException {
        StaticLog.info("全量同步： [" + tableName + "] 开始同步...");

        // 查询源表所有数据
        List<Entity> entities = sourceDb.query(SqlBuilder.create().select().from(tableName).build(), new EntityListHandler());
        StaticLog.info("全量同步：从源表 [" + tableName + "] 查询到 " + entities.size() + " 条记录。");

        //如果没有源表没有全量同步数据，则无须同步
        if(entities.size()==0){
            StaticLog.info("源表 [{}] 查询新增记录为0；无须同步", tableName);
            return;
        }

        // 删除目标表数据
        targetDb.execute(SqlBuilder.create().delete(tableName).build());
        StaticLog.info("全量同步：清空目标表 [" + tableName + "] 数据。");

        // 批量插入目标表数据
        batchInsert(targetDb, tableName, entities);

        StaticLog.info("全量同步：成功写入目标表 [" + tableName + "] 数据。");
        StaticLog.info("全量同步： [" + tableName + "] 同步完成");
    }

    /**
     * 增量同步（只处理新增数据）
     */
    private static void incrementalSync(Db sourceDb, Db targetDb, String tableName, String timestampField,String nowTimeStr) throws SQLException {
        //默认开始时间是当前时间
        String thisTalbeLastSyncTime = DateUtil.format(DateUtil.beginOfDay(DateUtil.date()), "yyyy-MM-dd HH:mm:ss");
        String selectLastTimeDateColumnSQL = "SELECT max("+timestampField+") as latestDateTime FROM "+ tableName ;

        //先去需要插入的数据中查询拿到最新数据的时间戳
        Entity lastTimeDateEntity = targetDb.queryOne(selectLastTimeDateColumnSQL);
        if(ObjUtil.isNotNull(lastTimeDateEntity.get("latestDateTime"))){
            thisTalbeLastSyncTime = lastTimeDateEntity.get("latestDateTime").toString();
        }

        StaticLog.info("表名:【"+tableName+"】-增量同步-同步时间：{}~{}", thisTalbeLastSyncTime,nowTimeStr);

        // 查询自上次同步时间之后新增的数据
        String sql = "SELECT * FROM "+ tableName + " WHERE "+timestampField +" BETWEEN '"+thisTalbeLastSyncTime+"' AND '"+nowTimeStr+"'" ;
        List<Entity> entities = sourceDb.query(sql, new EntityListHandler());
        StaticLog.info("增量同步：从源表 [{}] 查询到 {} 条新增记录。", tableName, entities.size());

        //如果没有增量同步数据，则无须同步
        if(entities.size()==0){
            StaticLog.info("源表 [{}] 查询新增记录为0；无须同步", tableName);
            return;
        }

        // 批量插入目标表数据
        batchInsert(targetDb, tableName, entities);
    }

    /**
     * 批量插入数据
     */
    private static void batchInsert(Db db, String tableName, List<Entity> entities) throws SQLException {
        // 获取表的所有字段名
        String[] fields = entities.get(0).getFieldNames().toArray(new String[0]); // 将 Set<String> 转换为 String[]

        // 构造批量插入 SQL (IGNORE代表忽略主键重复的数据)
        StringBuilder insertSQL = new StringBuilder("INSERT IGNORE INTO ")
                .append(tableName)
                .append(" (")
                .append(String.join(", ", fields))
                .append(") VALUES (")
                .append(repeatString("?", fields.length)) // 使用自定义方法生成占位符
                .append(")");

        // 使用 JDBC 批量插入
        try (Connection conn = db.getConnection();
             PreparedStatement ps = conn.prepareStatement(insertSQL.toString())) {

             conn.setAutoCommit(false); // 开启事务

            for (int i = 0; i < entities.size(); i++) {
                Entity entity = entities.get(i);

                // 设置参数
                for (int j = 0; j < fields.length; j++) {
                    ps.setObject(j + 1, entity.get(fields[j]));
                }

                ps.addBatch(); // 添加到批处理

                // 达到批量大小时执行批处理
                if ((i + 1) % batchSize == 0 || i == entities.size() - 1) {
                    ps.executeBatch(); // 执行批处理
                    ps.clearBatch(); // 清空批处理
                    StaticLog.info("批量插入：成功插入 {} 条记录到表 [{}]", Math.min(batchSize, i + 1), tableName);
                }
            }

            conn.commit(); // 提交事务
        } catch (SQLException e) {
            StaticLog.error("批量插入失败：", e);
            throw e;
        }
    }

    /**
     * 自定义字符串重复方法
     */
    private static String repeatString(String str, int count) {
        if (count <= 0) {
            return "";
        }
        StringBuilder result = new StringBuilder();
        for (int i = 0; i < count; i++) {
            result.append(str);
            if (i < count - 1) {
                result.append(",");
            }
        }
        return result.toString();
    }

}
